package kafka2Hbase.UserDefinedSource
import java.nio.charset.StandardCharsets
import java.util.Properties
import kafka2Hbase.bean.MyResult01
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.source.{RichParallelSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer

/**
 * @author HuBiWei
 * @create 2022/4/28
 */
class TestSource extends RichParallelSourceFunction[MyResult01] {


  var flag:Boolean=true
  var dataStream: DataStream[String] = null

  override def open(parameters: Configuration): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //消费kafka流进行处理
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "172.16.75.138:9092,172.16.75.139:9092,172.16.75.140:9092")
    properties.setProperty("group.id", "consumer-group")
    properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("auto.offset.reset", "latest")
    //FlinkKafkaConsumer011(topic: String, valueDeserializer: DeserializationSchema[T], props: Properties)
    dataStream= env.addSource(new FlinkKafkaConsumer[String]("s1",new SimpleStringSchema(StandardCharsets.UTF_8),properties))
  }

  override def run(ctx: SourceFunction.SourceContext[MyResult01]): Unit = {
    while(flag){

      val senseStream: DataStream[MyResult01] = dataStream.map(f=>{
        val arr = f.split(",")
        MyResult01(arr(0).trim,arr(1).trim.toInt)
      })
      val results = senseStream.executeAndCollect()
      while (results.hasNext) {
        val res = results.next()
        ctx.collect(new MyResult01(res.name,res.age))
      }
      }
    }

  override def cancel(): Unit = {
    flag=false
  }



}
